/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.pre.parser.put;

import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import cn.ac.iie.pre.parser.global.GlobalParas;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import jdk.nashorn.internal.objects.NativeArray;
import org.apache.log4j.Logger;
import org.apache.commons.configuration2.ex.ConfigurationException;
import org.apache.commons.lang.exception.ExceptionUtils;

/**
 *
 * @author gjw
 */
public class SendMQThread implements Runnable {

    static Logger LOG = Logger.getLogger(SendMQThread.class);
    String msgType;
    ArrayBlockingQueue<REMessage> msgQueue;
    SendMQ sendMQ;

    String[] colNames;
    REFieldType[] types;

    List<Object[]> dataList = new ArrayList<>();

    public SendMQThread(String msgType) {
        this.msgType = msgType;
        msgQueue = GlobalParas.SEND_QUEUE_MAP.get(this.msgType);
        this.sendMQ = null;
    }

    @Override
    public void run() {
        REMessage msg = null;

        long startTime = System.currentTimeMillis() / 1000;
        while (true) {
            try {
                msg = msgQueue.poll(2, TimeUnit.SECONDS);
            } catch (InterruptedException ex) {
                LOG.error(ExceptionUtils.getFullStackTrace(ex));
            }
            LOG.debug("sendMQ value: " + this.sendMQ);
            if (msg != null) {
                while (sendMQ == null) {
                    try {
                        this.sendMQ = new SendMQ(msg);
                        this.sendMQ.start();
                        this.colNames = msg.colNames;
                        this.types = msg.types;
//                    this.sendMQ.start();
                        LOG.info("Start sendMQ, msgType: " + this.msgType);
                    } catch (ConfigurationException ex) {
                        LOG.error(ExceptionUtils.getFullStackTrace(ex));
                    }
                }

                try {
                    addMapToSendDatas(msg);
                } catch (Exception ex) {
                    LOG.error(ExceptionUtils.getFullStackTrace(ex));
                }

            }
            try {
                int dataListSize = dataList.size();
                long endTime = System.currentTimeMillis() / 1000;
                if ((dataListSize >= GlobalParas.SEND_MQ_MSG_NUM || (endTime - startTime) >= GlobalParas.SEND_MQ_TIME_OUT) && dataListSize > 0) {
                    REMessage rEMessage = new REMessage(this.colNames, this.types, dataList, "", new HashMap<String, String>());
                    this.sendMQ.sendMessage(rEMessage);

                    dataList = new ArrayList<>();
                    startTime = endTime;
                }
            } catch (Exception e) {
                LOG.error(ExceptionUtils.getFullStackTrace(e));
            }

        }
    }

    private void addMapToSendDatas(REMessage message) {
        List<String> colNameList = Arrays.asList(message.colNames);
        List<Object[]> datas = message.data;
        LOG.debug(message.colNames.toString());
//        LOG.info(datas);
        for (Object[] data : datas) {
            LOG.debug(Arrays.asList(data));
            List<Object> tempDataList = new ArrayList<>();
            for (String column : colNameList) {
                int index = colNameList.indexOf(column);
                if (index < 0 || index >= colNameList.size()) {
                    LOG.error("index out of array, the index is: " + index);
                    continue;
                }
                Object obj = data[index];
                tempDataList.add(obj);
            }
            this.dataList.add(tempDataList.toArray());
        }
    }

}
